package kafka

import (
	"github.com/sirupsen/logrus"
	"go_knowledge/mq/kafka/event"
	"go_knowledge/mq/kafka/event/kafka/producer"
)

type defaultPusher struct {
	addr   []string
	logger *logrus.Logger
}

func NewDefaultPusher(addr []string) *defaultPusher {
	return &defaultPusher{addr: addr}
}

func (dp *defaultPusher) SetLogger(logger *logrus.Logger) {
	dp.logger = logger
}

func (dp *defaultPusher) SyncPush(event event.PbEvent, queueName, blockKey string) error {
	client := producer.NewDefaultSyncProducerClient(dp.addr, queueName)
	if nil != dp.logger {
		client.SetLogger(dp.logger)
	}

	// TODO 为了快速解决线上问题，临时解决方案,未来考虑AsyncProducer或者复用SyncProducer
	defer func() {
		client.Close()
	}()

	return client.PushPbEvent(event, blockKey)
}

func (dp *defaultPusher) Push(event event.PbEvent, queueName, blockKey string) {
	go func() {
		if err := dp.SyncPush(event, queueName, blockKey); nil != err {
			dp.logger.Errorf("push event failed: %+v", err)
		}
	}()
}
